package com.wykj.framework.spark;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.HashPartitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function0;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

import com.google.gson.Gson;
import com.wykj.framework.kafka.KafKaProducerUtil;

import kafka.serializer.StringDecoder;
import scala.Tuple2;

public class Demo {
	private static String SPARK_MASTER="local[4]";    //使用yarn作为资源框架
	private static String SPARK_DEPLOY_MODE="client";
	private static String SPARK_APP_NAME="sparkStreaming";
	private static long SPARK_BATCH_SIZE_M=10000;  //批处理间隔10s
	private static long SPARK_WIN_LEN=120000;   //窗口大小 2min （方便演示）
	private static long SPARK_WIN_SLID=10000;    //窗口滑动间隔10s 
	//checkpoint的路径
	//private static String SPARK_CHECKPOINT_DIR="hdfs://192.168.1.222:9000/spark/streaming/checkpoint4";

	private static String KAFKA_BOOTSTRAP_SERVERS="192.168.1.222:9092";
	private static String KAFKA_KEY_DESERIALIZER="org.apache.kafka.common.serialization.StringDeserializer";
	private static String KAFKA_VALUE_DESERIALIZER="org.apache.kafka.common.serialization.StringDeserializer";
	private static String KAFKA_GROUP_ID="test-consumer-group";
	private static String KAFKA_AUTO_OFFSET_RESET="largest";
	private static String KAFKA_ENABLE_AUTO_COMMIT="false";
	private static String KAFKA_TOPIC="test1";


	private static Function0<JavaStreamingContext> createStreamingContext=new Function0<JavaStreamingContext>() {

		public JavaStreamingContext call() throws Exception {
	
			SparkConf conf=new SparkConf();
			conf.set("spark.app.name", SPARK_APP_NAME);
			conf.set("spark.master", SPARK_MASTER);
			conf.set("spark.submit.deployMode", SPARK_DEPLOY_MODE);
			final JavaSparkContext jsc=new JavaSparkContext(conf);
			JavaStreamingContext jssc=new JavaStreamingContext(jsc,Duration.apply(SPARK_BATCH_SIZE_M));
			
			//checkpoint
			jssc.checkpoint("checkpoint");
			   
			//初始区域交易金额总和
			List<Tuple2<String,Long>> initRegionSumData=
			     Arrays.asList(new Tuple2<String,Long>("A01", Long.valueOf(500000)),new Tuple2<String,Long>("A02", Long.valueOf(8000000)),
			     new Tuple2<String,Long>("A03", Long.valueOf(15000000)),new Tuple2<String,Long>("A04", Long.valueOf(6000000)));
			   
			//一定要为JavaPairRDD类型
			JavaPairRDD<String,Long> initRegionSumRDD=jssc.sparkContext().parallelizePairs(initRegionSumData);
			  
			//初始交易总额
		    final long initAllSum=100000000;
		  
		    List<Tuple2<String,Long>> initAllSumData=Arrays.asList(new Tuple2<String,Long>("all",initAllSum));
		   
		    JavaPairRDD<String,Long> initAllSumRDD =jssc.sparkContext().parallelizePairs(initAllSumData);
		   
		    Map<String,String> kafkaParams=new HashMap<String,String>();
		    kafkaParams.put("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS);
		    kafkaParams.put("key.deserializer", KAFKA_KEY_DESERIALIZER);
		    kafkaParams.put("value.deserializer", KAFKA_VALUE_DESERIALIZER);
		    kafkaParams.put("group.id", KAFKA_GROUP_ID);
		    kafkaParams.put("auto.offset.reset", "smallest"/*KAFKA_AUTO_OFFSET_RESET*/);
		    kafkaParams.put("enable.auto.commit", KAFKA_ENABLE_AUTO_COMMIT);
		
		
		    Set<String> topics=new HashSet<String>();
		    topics.add(KAFKA_TOPIC);
		       
		    JavaPairInputDStream<String, String> kafkaS= KafkaUtils.createDirectStream(jssc, String.class,
		         String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics);
		       
		    //过虑key  从kafka中获取的消息 key对于我们说是没有用的
		    JavaDStream<String> jds= kafkaS.map(new Function<Tuple2<String,String>, String>() {
				public String call(Tuple2<String, String> v1) throws Exception {
				return v1._2();
				}
			});
			      
			//原始数据存入hbase中，源始报文使用tab制表符分割字段
			//格式  数据格式：交易流水号交易地区编码交易类型交易金额交易时间
			//例如： 00001A01T0120002017-08-01
			jds.foreachRDD(new VoidFunction<JavaRDD<String>>() {
		
				public void call(JavaRDD<String> t) throws Exception {
			
					//按照官网这里应该使用数据连接池，对每一个partition使用一个连接即可，后续优化
					t.foreachPartition(new VoidFunction<Iterator<String>>() {
			
						public void call(Iterator<String> t) throws Exception {
					
						   List<Put> listPut=new ArrayList<Put>();
						   while(t.hasNext()){
								String one=t.next();
								byte[] f=Bytes.toBytes("cf");
								String s[]=one.split("");
								ImmutableBytesWritable key=new ImmutableBytesWritable(Bytes.toBytes(s[0]));
								Put p=new Put(Bytes.toBytes(s[0]));
								p.addColumn(f, Bytes.toBytes("tradeId"), Bytes.toBytes(s[0]));
								p.addColumn(f, Bytes.toBytes("tradeAreaId"), Bytes.toBytes(s[1]));
								p.addColumn(f, Bytes.toBytes("tradeType"), Bytes.toBytes(s[2]));
								p.addColumn(f, Bytes.toBytes("tradeMoney"), Bytes.toBytes(s[3]));
								p.addColumn(f, Bytes.toBytes("tradeTime"), Bytes.toBytes(s[4]));
								listPut.add(p);
						   }
						   
						   Configuration conf=HBaseConfiguration.create();
						   conf.set("hbase.zookeeper.quorum", "fsl:2181");
						   HTable table=new HTable(conf, "trade");
						   table.put(listPut);
				           table.flushCommits();
				           table.close();
						}
					});
				}
			});
			    
			//返回<地区，交易金额>
			JavaPairDStream<String, Long> jpds= jds.mapToPair(new PairFunction<String, String, Long>() {

				public Tuple2<String, Long> call(String t) throws Exception {

				String[] s=t.split("");
			
				return new Tuple2(s[1],Long.valueOf(s[3]));
				}
			});
			     
			//多次使用  缓存起来
			jpds.cache();
			      
			//装换为key相同的交易记录（"all",金额）
			JavaPairDStream<String, Long> jpdsMap= jpds.mapToPair(new PairFunction<Tuple2<String,Long>, String, Long>() {
		
				public Tuple2<String, Long> call(Tuple2<String, Long> t) throws Exception {
					// TODO Auto-generated method stub
					return new Tuple2("all",t._2());
				}
			});
			     
			//计算总的交易金额，使用默认分区与初始值
			JavaPairDStream<String,Long> allResult= jpdsMap.updateStateByKey(new Function2<List<Long>, Optional<Long>, Optional<Long>>() {
			    //v1表示新进来的数据 v2表示当前的数据
				public Optional<Long> call(List<Long> v1, Optional<Long> v2) throws Exception {

					Long news=0L;
					if(v2.isPresent()){
						news=v2.get();
					}
					for(Long i : v1){
						news+=i;
					}
					return Optional.of(news);
				}
			},new HashPartitioner(jssc.sparkContext().defaultParallelism()), initAllSumRDD);
			     
			 //打印
			 System.out.println("======总交易金额====");
			 allResult.print();
			   
			 //发送到kafka中
			 sendMessageToKafka("topic_all",allResult);   
			   
			 //统计地区交易总和 使用initRegionSumRDD做为初始值
			 JavaPairDStream<String, Long> regionSum=jpds.updateStateByKey(new Function2<List<Long>, Optional<Long>, Optional<Long>>() {
		
				public Optional<Long> call(List<Long> v1, Optional<Long> v2) throws Exception {
					Long news=0L;
					if(v2.isPresent()){
						news=v2.get();
					}
					for(Long i : v1){
						news+=i;
					}
					return Optional.of(news);
				}
			}, new HashPartitioner(jssc.sparkContext().defaultParallelism()), initRegionSumRDD);
			   
			  //打印
			System.out.println("======地区交易总金额====");
			regionSum.print(1);
			sendMessageToKafka("topic_regionsum",regionSum);
			  
			  
			   //统计各个地区两年内的交易总额
			JavaPairDStream<String, Long> regionSumInTwoYears= jpds.reduceByKeyAndWindow(new Function2<Long, Long, Long>() {
				public Long call(Long v1, Long v2) throws Exception {
				// TODO Auto-generated method stub
				return v1+v2;
				}
			}, Duration.apply(SPARK_WIN_LEN), Duration.apply(SPARK_WIN_SLID));
			  
			System.out.println("======地区交易两年内总金额====");
			regionSumInTwoYears.print();     
		
			 
			sendMessageToKafka("topic_regionsumTwoYear",regionSumInTwoYears);
			 
			return jssc;
		}
	};


	//将消息发送到kafka中   以json格式发送
	private static void sendMessageToKafka(final String topic,JavaPairDStream<String, Long> ds){
		ds.foreachRDD(new VoidFunction<JavaPairRDD<String,Long>>() {

			public void call(JavaPairRDD<String, Long> t) throws Exception {
				List<Tuple2<String, Long>> list= t.collect();
			
			
				Map<String,String> message=new HashMap<String,String>();
		        for(Tuple2<String, Long> tp: list){
		        	message.put(tp._1(), String.valueOf(tp._2()));
		        }
		        
		        KafKaProducerUtil.sendMassage(topic,null,new Gson().toJson(message));
			}
		});
	}


	public static void main(String[] args) {
		System.setProperty( "hadoop.home.dir", "D:/hadoop-2.7.0" );
	    //这个做checkpoint 重启仍然能够回到最新一次的状态，还能保证数据不丢失
		JavaStreamingContext jssc=JavaStreamingContext.getOrCreate("checkpoint",createStreamingContext);
		jssc.start();
		try {
			jssc.awaitTermination();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

	}
}
